package com.nami.gate;

import com.nami.gate.api.service.RemoteDownMsgRecvService;
import com.nami.gate.api.service.RemoteLoginService;
import com.nami.gate.config.DubboConfig;
import com.nami.gate.config.GateConfig;
import com.nami.gate.config.NacosConfig;
import com.nami.gate.config.RpcConfig;
import com.nami.gate.config.rocketmq.RocketMqConfig;
import com.nami.gate.constants.CommonConst;
import com.nami.gate.dubbo.RemoteDownMsgRecvServiceImpl;
import com.nami.gate.dubbo.RemoteLoginServiceImpl;
import com.nami.gate.middleware.mq.MqConsumer;
import com.nami.gate.middleware.mq.MqSender;
import com.nami.gate.middleware.mq.impl.RocketMqConsumerImpl;
import com.nami.gate.middleware.mq.impl.RocketMqSenderImpl;
import com.nami.gate.network.NettyServerImpl;
import com.nami.gate.network.NetworkService;
import com.nami.gate.network.factory.NamedThreadFactory;
import com.nami.gate.process.upmsg.PackageProcessFactory;
import com.nami.gate.register.RegisterService;
import com.nami.gate.register.impl.NacosRegisterServiceImpl;
import com.nami.gate.register.request.RegisterRequest;
import org.apache.dubbo.config.*;
import org.apache.dubbo.config.bootstrap.DubboBootstrap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class NamiController {

    private static final Logger log = LoggerFactory.getLogger(NamiController.class);

    private GateConfig config;

    private MqSender sender;
    private MqConsumer consumer;
    private RegisterService registerService;
    private NetworkService networkService;
    private DubboBootstrap dubboBootstrap;

    public NamiController(GateConfig config) {
        this.config = config;

        // 获取mqconfig
        RocketMqConfig rocketMqConfig = config.getRocketMqConfig();

        // 初始化mq sender
        if (CommonConst.ROCKET_MQ.equals(config.getMqType())) {
            sender = new RocketMqSenderImpl(rocketMqConfig);

        }
        if (CommonConst.KAFKA.equals(config.getMqType())) {
            // 初始化kafka
        }

        // processFactory 放入里面new 会频繁new factory对象。调用init方法
        // 提至最上层
        networkService = new NettyServerImpl(config, sender, new PackageProcessFactory(sender, config.getRpcConfig()));

        // 方便日后 nami-web 客户端获取长连接接口 使用
        registerService = new NacosRegisterServiceImpl();

        RpcConfig rpcConfig = config.getRpcConfig();
        if (CommonConst.ROCKET_MQ.equals(rpcConfig.getConsumerMode())) {
            consumer = new RocketMqConsumerImpl(rocketMqConfig);
        } else {
            dubboBootstrap = DubboBootstrap.getInstance();
        }

    }


    public void start() throws Exception {
        startBasicService();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            NamiController.this.shutdown();
        }));

    }

    private void startBasicService() throws Exception {
        // 初始化mq sender
        sender.start();

        RpcConfig rpcConfig = config.getRpcConfig();
        if (CommonConst.ROCKET_MQ.equals(rpcConfig.getConsumerMode())) {
            consumer.start();
        } else {
            startDubbo();
        }

        // 将netty服务 注册到注册中心
        if (CommonConst.NACOS.equals(config.getRegisterType())) {
            RegisterRequest request = new RegisterRequest();
            NacosConfig nacosConfig = config.getNacosConfig();
            request.setServerAddr(nacosConfig.getServerAddr());
            request.setUsername(nacosConfig.getUsername());
            request.setPassword(nacosConfig.getPassword());
            request.setNamespace(nacosConfig.getNamespace());

            request.setIp(config.getIp());
            request.setPort(config.getGatePort());

            registerService.start(request);
        }

        // netty init
        networkService.start();
        log.info("[basic_service] init success.");
    }

    /**
     * dubbo 性能会下降
     */
    private void startDubbo() {
        DubboConfig dubboConfig = config.getDubboConfig();
        ApplicationConfig application = new ApplicationConfig();
//        application.setName("logic2gate-service");
        application.setName("gate-dubbo-service");
        application.setParameters(Collections.singletonMap("migration", "FORCE_APPLICATION"));

        // 线程池配置
        ExecutorService threadPool = new ThreadPoolExecutor(
            100, // 核心线程数
            200, // 最大线程数
            60L, // 线程空闲时间
            TimeUnit.SECONDS, // 时间单位
            new LinkedBlockingQueue<>(1000), // 阻塞队列
            new NamedThreadFactory("dubbo-server-") // 线程工厂
        );


        ProtocolConfig protocol = new ProtocolConfig();
        protocol.setName("dubbo");
//        protocol.setName("tri"); // 使用triple协议
//        protocol.setSerialization("protobuf");
//        protocol.setSerialization("fst");
        protocol.setPort(dubboConfig.getProtocolPort());

        protocol.setThreadpool("cached");
//        protocol.setThreadpool("limited"); // 使用自定义线程池
        protocol.setThreads(50); // 核心线程数
//        protocol.setThreadpool("fixed");


        RegistryConfig registry = new RegistryConfig();

        registry.setAddress(dubboConfig.getAddress());
        registry.setGroup("DUBBO_GROUP");
        registry.setPassword(dubboConfig.getUsername());
        registry.setUsername(dubboConfig.getPassword());
//        registry.setAddress("redis://192.168.223.21:6379");
//        registry.setUsername("dubbo");
//        registry.setPassword("123456");
//        registry.setGroup("DUBBO_GROUP");

        // 元数据中心配置
        MetadataReportConfig metadataReportConfig = new MetadataReportConfig();
        metadataReportConfig.setAddress(dubboConfig.getAddress());
        metadataReportConfig.setGroup("DEFAULT_GROUP");
        metadataReportConfig.setUsername(dubboConfig.getUsername());
        metadataReportConfig.setPassword(dubboConfig.getPassword());

        // 创建服务配置
        ServiceConfig<RemoteLoginService> service = new ServiceConfig<>();
        service.setInterface(RemoteLoginService.class);
        service.setRef(new RemoteLoginServiceImpl());
        service.setProtocol(protocol);
        service.setApplication(application);
        service.setRegistry(registry);
        service.setMetadataReportConfig(metadataReportConfig);
        service.export();

        ServiceConfig<RemoteDownMsgRecvService> remoteDownMsgRecvService = new ServiceConfig<>();
        remoteDownMsgRecvService.setInterface(RemoteDownMsgRecvService.class);
        remoteDownMsgRecvService.setRef(new RemoteDownMsgRecvServiceImpl());
        remoteDownMsgRecvService.setProtocol(protocol);
        remoteDownMsgRecvService.setApplication(application);
        remoteDownMsgRecvService.setRegistry(registry);
        remoteDownMsgRecvService.setMetadataReportConfig(metadataReportConfig);

        remoteDownMsgRecvService.setExecutor(threadPool); // 配置线程池

        remoteDownMsgRecvService.export();

        dubboBootstrap.application(application)
            .protocol(protocol)
            .registry(registry)
            .metadataReport(metadataReportConfig)
            .start();
//            .await();
    }

    public void shutdown() {
        log.info("will shutdown service.");

        // 1. 先发送s2c指令 让客户端断连 ==> 管理系统，发送转移其他服务指令

        // 2. 下线服务
        registerService.shutdown();
        networkService.shutdown();
        sender.shutdown();
        consumer.shutdown();

        if (dubboBootstrap != null) {
            dubboBootstrap.stop();
        }

    }


}
